/**
 * 
 */
package io.gitee.falllee.dao.core.table;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.calcite.linq4j.Linq4j;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import io.gitee.falllee.common.util.SnowflakeUtils;
import io.gitee.falllee.dao.core.HeaderBuilder;
import io.gitee.falllee.dao.enums.DbTypeEnum;
import io.gitee.falllee.dao.enums.KeyTypeEnum;
import io.gitee.falllee.dao.enums.SyncDatePrecisionEnum;
import io.gitee.falllee.dao.model.DaoConst;
import io.gitee.falllee.dao.model.DaoException;
import io.gitee.falllee.dao.persistence.DataMapper;
import io.gitee.falllee.dao.util.CommonUtils;
import io.gitee.falllee.dao.util.DaoUtils;

/**
 * @author batty
 * @date 2022-06-27
 *
 */
public class InsertBuilder extends HeaderBuilder {

    private static final String KEY_VALUE = "keyValue";
    
    private DataMapper dataMapper;
    
    private CacheBuilder cacheBuilder;
    
    private MessageBuilder messageBuilder;

    private String tableName;
    
    private String keyField;
    
    private Integer insertListMaxSize = 500;
    
    /**
     * 是否同步表（同步表的固定字段取实体传递的值）
     */
    private boolean syncTable;

    /**
     * 同步时间字段（如果syncTable=true，则可以指定此字段为同步时间字段，取数据库当前时间）
     */
    private String syncDateField;
    
    /**
     * 同步时间精度
     */
    private SyncDatePrecisionEnum syncDatePrecision;
    
    InsertBuilder(TableBuilder parent) {
        super(parent.getDaoConfig());
        this.dataMapper = parent.getDataMapper();
        this.cacheBuilder = parent.cacheBuilder();
        this.messageBuilder = parent.messageBuilder();
        this.tableName = parent.getTableName();
        this.keyField = parent.getKeyField();
        this.syncTable = parent.syncTable();
        this.syncDateField = parent.syncDateField();
        this.syncDatePrecision = parent.syncDatePrecision();
    }
    
    public InsertBuilder header(String key, String value) {
        if (headerParamsMap == null) {
            headerParamsMap = new HashMap<>(16);
        }
        headerParamsMap.put(key, value);
        return this;
    }
    
    public InsertBuilder insertListMaxSize(Integer insertListMaxSize) {
        this.insertListMaxSize = insertListMaxSize;
        return this;
    }

    public Boolean insert(Map<String, Object> t) {
        if (t == null) {
            return false;
        }
        Map<String, Object> map = new HashMap<>(16);
        setInsertFieldAndValue(t, map);
        // 非同步表组件级固定字段需要统一赋值;如果是同步表则使用实体传递的值
        if (!syncTable) {
            setEnableFlag(map, false);
            setCreateInfo(map);
            setUpdateInfo(map);
        } else {
            setSyncDate(map, syncDateField, syncDatePrecision.getPrecision());
        }
        boolean bl = insertDb(map, tableName);
        if (bl) {
            Object keyValue = map.get(DaoConst.KEY_VALUE);
            if (keyValue == null) {
                return false;
            }
            // 回写主键值
            t.put(keyField, keyValue);
            // 清除实体缓存
            cacheBuilder.clearCache(null);
            // 发送Kafka消息
            messageBuilder.sendMessage(keyValue, DaoConst.INSERT);
        }
        return bl;
    }

    public Boolean insertList(List<Map<String, Object>> list) {
        List<Map<String, Object>> tempList = Linq4j.asEnumerable(list).where(p -> !ObjectUtils.isEmpty(p)).toList();
        if (CollectionUtils.isEmpty(list)) {
            return false;
        }
        String keyType = daoConfig.getKeyType();
        if (DbTypeEnum.ORACLE.getDbType().equals(daoConfig.getDbType())) {
            return insertList4Oracle(tempList);
        }
        List<Map<String, Object>> listParam = getInsertValueList(tempList, keyType);
        Map<String, Object> map = new HashMap<>(16);
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        map.put(DaoConst.TABLE_NAME, tableName);
        // 非同步表组件级固定字段需要统一赋值;如果是同步表则使用实体传递的值
        if (!syncTable) {
            setEnableFlag(map, false);
            setCreateInfo(map);
            setUpdateInfo(map);
        } else {
            setSyncDate(map, syncDateField, syncDatePrecision.getPrecision());
        }
        int total = tempList.size();
        int maxSize = insertListMaxSize;
        int num = total / maxSize;
        int left = total % maxSize;
        if (left > 0) {
            num += 1;
        }
        Integer res = 0;
        for (int i = 0; i < num; i++) {
            List<Map<String, Object>> subList = CommonUtils.getSubList(listParam, maxSize, i);
            if (KeyTypeEnum.UUID.getType().equals(keyType) || KeyTypeEnum.SNOWFLAKE_ID.getType().equals(keyType)) {
                res += dataMapper.insertList(map, subList);
            } else {
                res += dataMapper.insertAutoList(map, subList);
            }
        }
        if (res > 0) {
            // 回写主键值到list
            List<Object> keyValueList = writeBackKeyValueList(list, listParam);
            // 清除实体缓存
            cacheBuilder.clearCache(null);
            // 发送Kafka消息
            messageBuilder.sendMessage(keyValueList, DaoConst.INSERT_LIST);
        }
        return res > 0;
    }

    private void setInsertFieldAndValue(Map<String, Object> t, Map<String, Object> map) {
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        // 记录主键字段及值
        map.put(DaoConst.KEY_FIELD, getQuoteField(keyField));
        Object keyValue = t.get(keyField);
        map.put(DaoConst.KEY_VALUE, keyValue);
        StringBuilder sbFields = new StringBuilder();
        StringBuilder sbValues = new StringBuilder();
        // 非同步表，业务固定字段从请求头设置
        if (!syncTable) {
            Map<String, Object> fixFieldValue = getFixedFieldValue();
            DaoUtils.setFixFieldValue(fixFieldValue, t);
        }
        List<String> configFields = getComponentConfigFields(syncTable, syncDateField);
        Iterator<Entry<String, Object>> iterator = t.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, Object> entry = iterator.next();
            String columnName = entry.getKey();
            Object objVal = entry.getValue();
            String colQuotes = getQuoteField(columnName);
            // 空值或主键不处理(在sql中组装)
            if (ObjectUtils.isEmpty(objVal) || keyField.equals(columnName) || configFields.contains(columnName)) {
                continue;
            }
            // 组装field
            sbFields.append(colQuotes);
            sbFields.append(DaoConst.COMMA);
            // 组装value(参数化)
            sbValues.append(String.format("#{%s}", columnName));
            sbValues.append(DaoConst.COMMA);
            // 赋值
            map.put(columnName, objVal);
        }
        if (sbFields.length() > 0) {
            String fields = sbFields.substring(0, sbFields.length() - 1);
            map.put(DaoConst.FIELDS, fields);
        }
        if (sbValues.length() > 0) {
            String values = sbValues.substring(0, sbValues.length() - 1);
            map.put(DaoConst.VALUES, values);
        }
    }

    private boolean insertDb(Map<String, Object> map, String tableName) {
        Integer res = 0;
        map.put(DaoConst.TABLE_NAME, tableName);
        String keyType = daoConfig.getKeyType();
        map.put(DaoConst.KEY_TYPE, keyType);
        if (keyType.equals(KeyTypeEnum.UUID.getType())) {
            res = dataMapper.insertUuid(map);
        } else if (KeyTypeEnum.SNOWFLAKE_ID.getType().equals(keyType)) {
            // 如果主键值为空再填充雪花ID
            if (ObjectUtils.isEmpty(map.get(DaoConst.KEY_VALUE))) {
                long id = SnowflakeUtils.nextId();
                map.put(DaoConst.KEY_VALUE, id);
            }
            res = dataMapper.insert(map);
        } else {
            if (daoConfig.getDbType().equals(DbTypeEnum.ORACLE.getDbType())) {
                res = dataMapper.insertAuto4Oracle(map);
            } else {
                res = dataMapper.insertAuto(map);
            }
        }
        return res > 0;
    }

    private List<Map<String, Object>> getInsertValueList(List<Map<String, Object>> list, String keyType) {
        List<Map<String, Object>> listParam = new ArrayList<>();
        Map<String, Object> fixFieldValue = getFixedFieldValue();
        List<String> configFields = getComponentConfigFields(syncTable, syncDateField);
        for (int i = 0; i < list.size(); i++) {
            Map<String, Object> t = list.get(i);
            if (!syncTable) {
                DaoUtils.setFixFieldValue(fixFieldValue, t);
            }
            // 先设置主键值
            Object keyValue = t.get(keyField);
            keyValue = DaoUtils.generateKeyValue(keyType, keyValue);
            t.put(keyField, keyValue);
            Map<String, Object> mapValues = new HashMap<>(16);
            Iterator<Entry<String, Object>> iterator = t.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, Object> entry = iterator.next();
                String columnName = getQuoteField(entry.getKey());
                // 配置字段SQL层处理
                if (configFields.contains(columnName)) {
                    continue;
                }
                Object objVal = entry.getValue();
                mapValues.put(columnName, objVal);
            }
            listParam.add(mapValues);
        }
        return listParam;
    }

    private List<Object> writeBackKeyValueList(List<Map<String, Object>> list, List<Map<String, Object>> listParam) {
        if (CollectionUtils.isEmpty(list)) {
            return new ArrayList<>();
        }
        List<Object> keyValueList = new ArrayList<>();
        for (int i = 0; i < list.size(); i++) {
            Map<String, Object> mapValue = listParam.get(i);
            Object keyValue = mapValue.get(KEY_VALUE);
            if (keyValue != null) {
                Map<String, Object> map = list.get(i);
                map.put(keyField, keyValue);
                keyValueList.add(keyValue);
            }
        }
        return keyValueList;
    }

    private Boolean insertList4Oracle(List<Map<String, Object>> list) {
        Map<String, Object> map = new HashMap<>(16);
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        String keyType = daoConfig.getKeyType();
        setFieldAndValue(list.get(0), map);
        List<Object> keyValueList = setFieldAndValueList(list, keyType);
        map.put(DaoConst.TABLE_NAME, tableName);
        if (!syncTable) {
            setEnableFlag(map, false);
            setCreateInfo(map);
            setUpdateInfo(map);
        } else {
            setSyncDate(map, syncDateField, syncDatePrecision.getPrecision());
        }
        int total = list.size();
        int maxSize = insertListMaxSize;
        int num = total / maxSize;
        int left = total % maxSize;
        if (left > 0) {
            num += 1;
        }
        Integer res = 0;
        for (int i = 0; i < num; i++) {
            List<Map<String, Object>> subList = CommonUtils.getSubList(list, maxSize, i);
            res += dataMapper.insertList4Oracle(map, subList);
        }
        if (res > 0) {
            // 清除实体缓存
            cacheBuilder.clearCache(null);
            // 发送Kafka消息
            messageBuilder.sendMessage(keyValueList, DaoConst.INSERT_LIST);
        }
        return res > 0;
    }

    private void setFieldAndValue(Map<String, Object> t, Map<String, Object> map) {
        StringBuilder sbFields = new StringBuilder();
        StringBuilder sbValues = new StringBuilder();
        List<String> configFields = getComponentConfigFields(syncTable, syncDateField);
        Iterator<Entry<String, Object>> iterator = t.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, Object> entry = iterator.next();
            String columnName = entry.getKey();
            String colQuotes = getQuoteField(columnName.toUpperCase());
            if (sbFields.toString().contains(colQuotes) || configFields.contains(columnName)) {
                continue;
            }
            // 组装field
            sbFields.append(colQuotes);
            sbFields.append(DaoConst.COMMA);
            // 组装value(参数化)
            sbValues.append(String.format("#{item.%s}", columnName));
            sbValues.append(DaoConst.COMMA);
        }
        String fields = "";
        String values = "";
        if (sbFields.length() > 0) {
            fields = sbFields.substring(0, sbFields.length() - 1);
            map.put("fields", fields);
        } else {
            throw new DaoException(DaoConst.NO_FIELDS_MAPPING);
        }
        if (sbValues.length() > 0) {
            values = sbValues.substring(0, sbValues.length() - 1);
            // 注:这里key不能用values,values为map的属性
            map.put("vals", values);
        }
    }
    
    private List<Object> setFieldAndValueList(List<Map<String, Object>> list, String keyType) {
        Map<String, Object> fixFieldValue = getFixedFieldValue();
        List<Object> keyValueList = new ArrayList<>();
        for (int i = 0; i < list.size(); i++) {
            Map<String, Object> t = list.get(i);
            // 非同步表设置固定字段的值
            if (!syncTable) {
                DaoUtils.setFixFieldValue(fixFieldValue, t);
            }
            Iterator<Entry<String, Object>> iterator = t.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, Object> entry = iterator.next();
                String columnName = entry.getKey();
                Object objVal = entry.getValue();
                if (keyField.equals(columnName)) {
                    // 如果主键值为空且主键类型为UUID的自动生成主键,自增主键清空主键值，数据层自动填充
                    objVal = getOracleKeyValue(tableName, keyType, objVal);
                    entry.setValue(objVal);
                    keyValueList.add(objVal);
                }
            }
        }
        return keyValueList;
    }

    private Object getOracleKeyValue(String tableName, String keyType, Object objVal) {
        if (ObjectUtils.isEmpty(objVal)) {
            if (KeyTypeEnum.UUID.getType().equals(keyType)) {
                objVal = CommonUtils.getGuid();
            } else if (KeyTypeEnum.SNOWFLAKE_ID.getType().equals(keyType)) {
                objVal = SnowflakeUtils.nextId();
            }
        } else if (keyType.equals(KeyTypeEnum.AUTO.getType())) {
            String sequence = String.format("%s_S", tableName);
            objVal = dataMapper.getOracleSerialNumber(sequence);
        }
        return objVal;
    }
}
